package com.fanli.bigdata.mytest

import com.fanli.bigdata.db.StateRedisConnections

import scala.collection.mutable.ArrayBuffer
import java.text.SimpleDateFormat
import java.util.{Date, Locale}
import org.json4s.jackson.JsonMethods._
import org.joda.time.format.DateTimeFormat

object GoshopFun {
  def MatchLogBatchFunc(msgs: Iterator[String]): Iterator[(RealtimeGoShopPUvKey,RealtimeGoShopPUvValue)] = {
    val arr = ArrayBuffer[(RealtimeGoShopPUvKey,RealtimeGoShopPUvValue)]()
    msgs.foreach(msg=>{
      var goshop  =getLogFunc(msg)
      var uvSet = Set[String]()
      var pvSet = Set[String]()
      if(goshop.uid !=""){
        uvSet += goshop.uid
      }
      pvSet += goshop.tracking_id
      if (goshop.pid != null && "300004".equals(goshop.serial_id)) {
        arr += ((RealtimeGoShopPUvKey(goshop.ds,goshop.hour,goshop.pid),RealtimeGoShopPUvValue(pvSet,uvSet)))
      }
    })
    arr.toIterator
  }

  def MatchLogSteamingFunc(msgs: Iterator[(String, String)]): Iterator[(RealtimeGoShopPUvKey,RealtimeGoShopPUvValue)] = {
    val arr = ArrayBuffer[(RealtimeGoShopPUvKey,RealtimeGoShopPUvValue)]()
    msgs.foreach(msg=>{
      var goshop  =getLogFunc(msg._2)
      var uvSet = Set[String]()
      var pvSet = Set[String]()
      if(goshop.uid !=""){
        uvSet += goshop.uid
      }
      pvSet += goshop.tracking_id
      if (goshop.pid != null && "300004".equals(goshop.serial_id)) {
        arr += ((RealtimeGoShopPUvKey(goshop.ds,goshop.hour,goshop.pid),RealtimeGoShopPUvValue(pvSet,uvSet)))
      }
    })
    arr.toIterator
  }

  def goshopState(res: Array[(RealtimeGoShopPUvKey,RealtimeGoShopPUvValue)]):Unit={
    val redisClient = new StateRedisConnections("115.159.45.213",6379)
    val redis = redisClient.redis
    val REDIS_EXPIR: Int = 24 * 60 * 60 * 2 //2day

    //    var url = "jdbc:mysql://115.159.45.213:3306/test?useUnicode=true&characterEncoding=utf8"
    //    val conn= DriverManager.getConnection(url,"hive","hive")
    //    var stmt = conn.createStatement()
    res.foreach(r=>{
      val prefix_key = "bi_goshp_" + r._1.pid + r._1.ds + "_"
      val prefix_detail_key = prefix_key + r._1.hour + "_"

      val total_pvs = r._2.pv.toSeq
      val total_uvs = r._2.us.toSeq

      val prefix_key_pv = prefix_key + "pv"
      val prefix_key_uv = prefix_key + "uv"
      //日 pv
      redis.pfadd(prefix_key_pv, total_pvs(0), total_uvs: _*)
      redis.expire(prefix_key_pv,REDIS_EXPIR)
      //日 uv
      redis.pfadd(prefix_key_uv, total_uvs(0), total_uvs: _*)
      redis.expire(prefix_key_uv,REDIS_EXPIR)

      val prefix_detail_key_pv = prefix_detail_key + "pv"
      val prefix_detail_key_uv = prefix_detail_key + "uv"
      //小时 pv
      redis.pfadd(prefix_detail_key_pv, total_uvs(0), total_uvs: _*)
      redis.expire(prefix_detail_key_pv,REDIS_EXPIR)
      //小时 uv
      redis.pfadd(prefix_detail_key_uv, total_uvs(0), total_uvs: _*)
      redis.expire(prefix_detail_key_uv,REDIS_EXPIR)

      val  dpvCount = redis.pfcount(prefix_key_pv)
      val  duvCount = redis.pfcount(prefix_key_uv)
      val  hpvCount = redis.pfcount(prefix_detail_key_pv)
      val  huvCount = redis.pfcount(prefix_detail_key_uv)

      println(dpvCount + "\t" + duvCount +"\t"+ hpvCount + "\t" + huvCount)

      //      val dSql = "INSERT INTO tb_varnish_realtime_flow(record_date, pv, uv, create_time) VALUES(\"" +
      //        r._1.ds + "\"," + dpvCount + "," + duvCount + ", NOW() )"+
      //        "ON DUPLICATE KEY UPDATE pv=" + dpvCount + ",uv=" + duvCount
      //      println(dSql)
      //      stmt.addBatch(dSql)
    })
    //    stmt.executeBatch()
    //    stmt.close()
    //    conn.close
  }

  def getLogFunc(log:String): RealtimeGoShop = {
    val log_bak = """192.168.2.206 data={"data1":6444967106,"data2":1478098831,"data3":"23995644|23995644|23995644","data10":"super_women_brand|600148|super","data12":"539033434659","serial_id":300004,"time":1478098831,"user_id":"23995644","ip":"112.14.84.88"} 1532 2016-11-02 23:00:31"""
//    val log = """192.168.2.206 data={"data1":6444967106,"data2":1478098831,"data3":"","data10":"super_women_brand","data12":"539033434659","serial_id":300004,"time":1478098831,"user_id":"23995644","ip":"112.14.84.88"} 1532 2016-11-02 23:00:31""".stripMargin
    val start = log.indexOf("data={")
    val end = log.lastIndexOf("}")
    val jsonData = log.substring(start + 5, end + 1)
    val json = parse(jsonData)

    val serial_id = json \ "serial_id" values
    val tracking_id = json \ "data1" values
    var pid = json \ "data12" values
    var lc_JV = json \ "data10" values
    val userid_jason = json \ "data3" values
    val time = json \ "time" values

    val ymdSdf = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm").withLocale(Locale.US)

    var lc = lc_JV.toString.split("\\|")(0)

    val ymdhTime = ymdSdf.print(time.toString().toLong * 1000).split(" ")
    val ds = ymdhTime(0)
    val hour = ymdhTime(1).split(":")(0)

    var uid = ""
    if (userid_jason != null && userid_jason != "") {
      val splitData3 = userid_jason.toString.split("\\|")
      if (splitData3(2) != null && splitData3(2) != "") {
        uid = splitData3(2)
      } else if (splitData3(1) != null && splitData3(1) != "") {
        uid = splitData3(1)
      } else if (splitData3(0) != null && splitData3(0) != "") {
        uid = splitData3(0)
      }
    }
    RealtimeGoShop(ds,hour,tracking_id.toString,serial_id.toString,pid.toString,uid,lc)
  }
}
